package main.java.hotProd

import cn.hutool.core.date.DateUtil
import cn.hutool.core.util.StrUtil
import org.apache.log4j.Logger
import org.apache.log4j.Level
import org.apache.spark.{SparkConf, SparkContext}

/**
  * HotProdDataClean
  *
  * @author zhangyimin
  * @date 2018-11-09 下午5:47
  * @version 1.0
  */
object HotProdDataClean {

  def main(args: Array[String]): Unit = {

    Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)
    Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)


    //正常数据 1,201.105.101.102,http://mystore.jsp/?productid=1,2017020029,2,1
    val conf = new SparkConf().setAppName("hot_prod_clean_data").setMaster("local")
    val sc = new SparkContext(conf)

    val dataRdd = sc.textFile("hdfs://10.16.7.36:9000/flume/**")

    val line = dataRdd.map(_.split(","))
    //使用MAP
    val res = line.map(x => {
      if (x.length == 6 && StrUtil.isNotBlank(x(2))) {
        //数据转换成字符,并以逗号隔开
        //        println(x(3))
        //        ArrayUtil.join(x,",")
        x
      } else {
        null
      }
    })
    res.filter(x => x != null && x.length > 0).map(x => x.mkString(",")).saveAsTextFile("hdfs://10.16.7.36:9000/data/output/hot_products/data_clean/out_" + DateUtil.current(false))


    //使用filter
    //    val res=line.filter(x=>x.length==6&&StrUtil.isNotBlank(x(2)))
    //    res.map(x=>x.mkString(",")).saveAsTextFile("hdfs://10.16.7.36:9000/data/output/hot_products/data_clean/out_" + DateUtil.current(false))


    sc.stop


  }

}
